package com.shujia.spark.mllib

import org.apache.spark.ml.linalg.Vectors
import org.apache.spark.sql.{DataFrame, Row, SparkSession}

object Demo3ImageMake {
  def main(args: Array[String]): Unit = {
    val spark: SparkSession = SparkSession
      .builder()
      .master("local[8]")
      .appName("image")
      .getOrCreate()
    import spark.implicits._

    //读取图片数据
    val imageData: DataFrame = spark.read
      .format("image")
      .load("C:\\data\\train")

    imageData.printSchema()
    /**
     * 1、特征工程：将原始的数据转换成算法可以识别的向量
     */
    //处理图片数据
    val featuresDF: DataFrame = imageData
      .select($"image.origin", $"image.data")
      .map {
        case Row(origin: String, date: Array[Byte]) =>
          val array: Array[Double] = date
            .map(_.toInt)
            //归一化处理
            .map(i => if (i < 0) 1.0 else 0.0)

          //取出文件名
          val name: String = origin.split("/").last

          //转换成向量返回
          (name, Vectors.dense(array).toSparse)
      }.toDF("name", "features")

    //读取标记文件
    val labelDF: DataFrame = spark.read
      .format("csv")
      .option("sep", " ")
      .schema("name STRING,label DOUBLE")
      .load("data/train.txt")

    //关联目标值和特征向量
    val dataDF: DataFrame = featuresDF
      .join(labelDF, "name")
      .select($"label", $"features")

    //保存数据
    dataDF
      .write
      .format("libsvm")
      .save("data/image_data")
  }
}
